package cn.ycc1.boot3kafka.service;

import cn.ycc1.boot3kafka.entity.UserMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

/**
 * @author ycc
 * @date 2025/11/16
 */
@Service
public class KafkaProducerService {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Value("${app.kafka.topic.demo}")
    private String demoTopic;

    @Value("${app.kafka.topic.user}")
    private String userTopic;

    /**
     * 发送简单消息
     */
    public void sendMessage(String message) {
        CompletableFuture<SendResult<String, String>> future =
                kafkaTemplate.send(demoTopic, message);

        future.whenComplete((result, ex) -> {
            if (ex == null) {
                System.out.println("发送消息成功: " + message +
                        ", offset: " + result.getRecordMetadata().offset());
            } else {
                System.err.println("发送消息失败: " + ex.getMessage());
            }
        });
    }

    /**
     * 发送带key的消息
     */
    public void sendMessageWithKey(String key, String message) {
        kafkaTemplate.send(demoTopic, key, message)
                .whenComplete((result, ex) -> {
                    if (ex == null) {
                        System.out.println("发送带key消息成功: key=" + key +
                                ", value=" + message);
                    } else {
                        System.err.println("发送带key消息失败: " + ex.getMessage());
                    }
                });
    }

    /**
     * 发送用户消息
     */
    public void sendUserMessage(UserMessage userMessage) {
        try {
            // 这里可以使用 Jackson 将对象转为 JSON
            String message = String.format("{\"id\":%d,\"name\":\"%s\",\"email\":\"%s\"}",
                    userMessage.getId(), userMessage.getName(), userMessage.getEmail());

            kafkaTemplate.send(userTopic, message)
                    .whenComplete((result, ex) -> {
                        if (ex == null) {
                            System.out.println("发送用户消息成功: " + userMessage);
                        } else {
                            System.err.println("发送用户消息失败: " + ex.getMessage());
                        }
                    });
        } catch (Exception e) {
            System.err.println("序列化用户消息失败: " + e.getMessage());
        }
    }
}